package HandshakeJoin;

import java.util.List;
import java.net.URL;
import com.google.common.collect.Lists;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class InputConfigure
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(InputConfigure.class);

    private static String STREAM_R = "ExtDataR";
    private static String STREAM_S = "ExtDataS";

    private String sourceFilename = null;
    private int emitInterval = 50;
    private int sizeOfOneEmit = 1;
    private int numFields = 2;
    private boolean randomSource = true;

    public InputConfigure() {}

    public InputConfigure parseInputDespFile(final URL url,
            final String streamName) {
        Preconditions.checkArgument(streamName.equals(STREAM_R)
                || streamName.equals(STREAM_S));

        List<String> lines = Configure.getLines(url);

        parseEntries(lines, streamName);

        return this;
    }

    private void parseEntries(List<String> lines, final String streamName) {
        String arg;
        do {
            Configure.removeLinesTillSeparatorBar(lines);
            if (lines.isEmpty()) {
                break;
            }
            else {
                Configure.removeLinesTillNonempty(lines);
                arg = Configure.parseOneLine(lines.get(0));
                System.out.println("Stream name: " + arg);
            }
        }
        while (!arg.equals(streamName));
        Preconditions.checkState(!lines.isEmpty(),
                "no matching configuration set");

        parseEntriesForInputAdapter(lines);
    }

    private void parseEntriesForInputAdapter(List<String> entries) {
        final int numEntriesToParse = 6;

        List<String> args = Lists.newArrayListWithCapacity(numEntriesToParse);
        for (int i = 0; i < numEntriesToParse; ++i) {
            args.add(Configure.parseOneLine(entries.get(i)));
        }

        int i = 1; // skip the Stream Name
        sourceFilename = args.get(i++);
        emitInterval = Integer.parseInt(args.get(i++));
        sizeOfOneEmit = Integer.parseInt(args.get(i++));
        randomSource = Boolean.parseBoolean(args.get(i++));
        numFields = Integer.parseInt(args.get(i++));
    }

    public String getSourceFilename() {
        return sourceFilename;
    }

    public int getEmitInterval() {
        return emitInterval;
    }

    public int getSizeOfOneEmit() {
        return sizeOfOneEmit;
    }

    public boolean getEnableRandomSource() {
        return randomSource;
    }

    public int getNumberOfFieldsPerTuple() {
        return numFields;
    }
}